package com.gtp.kafka.demo;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class Demo2 {

	private final static String TOPIC = "markTopic";
    private final static String BOOTSTRAP_SERVERS = "10.103.27.42:9091";

    private static Consumer<Long, String> createConsumer() {
        final Properties props = new Properties();
        
        //服务列表  组  序列化
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaExampleConsumer");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        final Consumer<Long, String> consumer = new KafkaConsumer<>(props);

        return consumer;
    }
    
    public static void getPartitionsForTopic() {
        final Consumer<Long, String> consumer = createConsumer();
        
        Collection<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC);
        
        List<TopicPartition> tp =new ArrayList<TopicPartition>();
       
        partitionInfos.forEach(p -> {
            
            tp.add(new TopicPartition(TOPIC,p.partition()));
           
            consumer.assign(tp); //自己分配
            consumer.seekToEnd(tp); //设置到最后

            System.out.println(p.partition() 
            			+ "->" + 
            			consumer.position(new TopicPartition(TOPIC, p.partition())));
        });
    }
    
    //查询topic的最新offset
    public static void main(String[] args) {
    	getPartitionsForTopic();
    }
}